package job

import (
	"git.multiverse.io/eventkit/common/model/mgmt"
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/dxc-server/constant"
	"git.multiverse.io/eventkit/dxc-server/mgmt/abnormal_txn"
	"git.multiverse.io/eventkit/dxc-server/mgmt/redundancy_abnormal_txn"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/eventkit/dxc-server/repo"
	"git.multiverse.io/framework/log"
	"reflect"
	"sort"
	"strings"
)

func StartRedundancyAbnormalTransactionJob() {
	repoType := config.Cfg.Repo.Type
	if repoType != repo.Hybrid_With_Redis_Storage && repoType != repo.Hybrid_With_Hazelcast_Storage {
		log.Debugsf("Repository is %s, no need to get redundancy abnormal transactions", repoType)
		return
	}
	if !config.Cfg.Management.Enable {
		log.Warns("DTS management has disabled, skip...")
		return
	}
	txnRepo := repo.RepositoryInstance()
	hybridRepo := txnRepo.(*repo.HybridStorage)
	go redundancyAbnormalTransactionWorker(hybridRepo)
}

func redundancyAbnormalTransactionWorker(hybridRepo *repo.HybridStorage) {
	memoryRepo := hybridRepo.GetMemoryRepo()
	nodeId := config.Cfg.NodeId
	batchSize := int(config.Cfg.Job.BatchSyncSize)
	// query page abnormal transaction
	var err error
	records := make([]*mgmt.PageQueryRedundancyAbnormalTxnRecordResult, 0)
	page := 1
	for {
		transactions, e := redundancy_abnormal_txn.PageQueryRedundancyAbnormalTransactions(nodeId, page, batchSize)
		if e != nil {
			err = e
			break
		}
		if len(transactions) == 0 {
			break
		}
		records = append(records, transactions...)
	}
	if err != nil {
		log.Errorsf("")
	}
	// delete contain DeleteTxn method
	results := groupByRootXid(records)
	mp := make(map[int]bool)
	for index, result := range results {
		hasDelete := false
		for _, record := range result {
			if record.MethodName == "DeleteTxn" {
				hasDelete = true
			}
		}
		mp[index] = hasDelete
	}
	abnormalRecords := make([][]*mgmt.PageQueryRedundancyAbnormalTxnRecordResult, 0)
	for index, result := range results {
		if !mp[index] {
			abnormalRecords = append(abnormalRecords, result)
		}
	}
	// operate abnormal transaction
	for _, transactions := range abnormalRecords {
		for _, transaction := range transactions {
			if transaction.GlobalTxnJSON != "" {
				origin := new(models.RootTransaction)
				_ = json.Unmarshal([]byte(transaction.GlobalTxnJSON), origin)
				memoryRepo.Recovery(origin)
			}
			// invoke abnormal method
			invokeAbnormalMethod(memoryRepo, transaction)
			root, _ := memoryRepo.FindAllBranchByRootXid(transaction.RootXId)
			reason := "redundancy abnormal transaction fail"
			abnormal_txn.ReportTransactionToDTSManagement(root, reason)
			memoryRepo.DeleteTxn(transaction.RootXId)
		}
	}
}

type sortByCreateTime []*mgmt.PageQueryRedundancyAbnormalTxnRecordResult

func (s sortByCreateTime) Len() int           { return len(s) }
func (s sortByCreateTime) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
func (s sortByCreateTime) Less(i, j int) bool { return s[i].CreateTime.Before(s[j].CreateTime) }

func groupByRootXid(list []*mgmt.PageQueryRedundancyAbnormalTxnRecordResult) [][]*mgmt.PageQueryRedundancyAbnormalTxnRecordResult {
	sort.Sort(sortByCreateTime(list))
	returnData := make([][]*mgmt.PageQueryRedundancyAbnormalTxnRecordResult, 0)
	i := 0
	var j int
	for {
		if i >= len(list) {
			break
		}
		for j = i + 1; j < len(list) && list[i].RootXId == list[j].RootXId; j++ {
		}
		returnData = append(returnData, list[i:j])
		i = j
	}
	return returnData
}

func invokeAbnormalMethod(repo repo.Repository, transaction *mgmt.PageQueryRedundancyAbnormalTxnRecordResult) error {
	instance := reflect.ValueOf(repo)
	fieldJsons := strings.Split(transaction.MethodFieldsJSON, constant.REDUNDANCY_ABNORMAL_TRANSACTION_METHOD_FIELDS_JSON_SPLIT)

	method := reflect.ValueOf(instance).MethodByName(transaction.MethodName)
	pTypes := make([]reflect.Type, 0)
	for i := 0; i < method.Type().NumIn(); i++ {
		pTypes = append(pTypes, method.Type().In(i))
	}
	pValues := make([]reflect.Value, 0)
	for i, pType := range pTypes {
		var mValue reflect.Value
		if pType.Kind() == reflect.Ptr {
			mValue = reflect.New(pType.Elem())
		} else {
			mValue = reflect.New(pType)
		}
		if err := json.Unmarshal([]byte(fieldJsons[i]), mValue.Interface()); err != nil {
			return err
		}
		if pType.Kind() == reflect.Ptr {
			pValues[i] = mValue
		} else {
			pValues[i] = mValue.Elem()
		}
	}
	method.Call(pValues)
	return nil
}
